package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.dto.ArticleStreamMessage;
import com.heima.article.dto.UpdateArticleMessage;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;

@EnableBinding(IHotArticleProcessor.class)
public class HotArticleListener {

    @StreamListener("hot_article_score_topic")
    @SendTo("hot_article_result_topic")
    public KStream<String, String> process(KStream<String, String> input) {
        //        topic最原本的数据

        KStream<String, String> kStream = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                return new KeyValue<>(updateArticleMessage.getArticleId().toString(), value);
            }
        });



        // 根据key分组
//      现在的数据：
//        {"add":1,"articleId":1543159543675031553,"type":2}
//        {"add":1,"articleId":1543159543675031553,"type":2}
//        {"add":1,"articleId":1543159543675031553,"type":1}
//        {"add":1,"articleId":1543159543675031553,"type":1}
//        {"add":1,"articleId":1543159543675031553,"type":3}

//        {"add":1,"articleId":1542829220378542082,"type":1}
//        {"add":1,"articleId":1542829220378542082,"type":1}
//        {"add":1,"articleId":1542829220378542082,"type":0}
//        {"add":1,"articleId":1542829220378542082,"type":1}
//        {"add":1,"articleId":1542829220378542082,"type":3}
        KGroupedStream<String, String> kGroupedStream = kStream.groupByKey();

        // 根据时间窗口统计  每个60秒
        TimeWindowedKStream<String, String> windowedKStream = kGroupedStream.windowedBy(TimeWindows.of(Duration.ofSeconds(60)));

        //        聚合之前的初始化器 内容为空
        Initializer<String> initializer = new Initializer<String>() {
            @Override
            public String apply() {
                return null;
            }
        };
        //        开始聚合
        Aggregator<String,String,String> aggregator = new Aggregator<String, String, String>() {
            @Override
            public String apply(String key, String value, String aggregate) {
                ArticleStreamMessage articleStreamMessage = null;
                if (aggregate == null){
                    articleStreamMessage = new ArticleStreamMessage();
                    articleStreamMessage.setArticleId(Long.parseLong(key));
                    articleStreamMessage.setView(0);
                    articleStreamMessage.setCollect(0);
                    articleStreamMessage.setComment(0);
                    articleStreamMessage.setLike(0);
                }else {
                    articleStreamMessage = JSON.parseObject(aggregate, ArticleStreamMessage.class);
                }

                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                Integer type = updateArticleMessage.getType();

                switch (type){
                    case 0:  //阅读
                        articleStreamMessage.setView(articleStreamMessage.getView()+updateArticleMessage.getAdd());
                        break;
                    case 1: //点赞
                        articleStreamMessage.setLike(articleStreamMessage.getLike()+updateArticleMessage.getAdd());
                        break;
                    case 2: //评论
                        articleStreamMessage.setComment(articleStreamMessage.getComment()+updateArticleMessage.getAdd());
                        break;
                    case 3: //收藏
                        articleStreamMessage.setCollect(articleStreamMessage.getCollect()+updateArticleMessage.getAdd());
                        break;
                }


                //key:文章id
                // value：UpdateArticleMessage{"add":1,"articleId":1456088799662469121,"type":2}
                // aggregate聚合后的结果 {articleId:"1456088799662469121",view:1 , like:1, comment:1 , collect:2 }
                return JSON.toJSONString(articleStreamMessage);
            }
        };

        // 聚合
        KTable<Windowed<String>, String> kTable = windowedKStream.aggregate(initializer, aggregator);

        KStream<String, String> kStream1 = kTable.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                return new KeyValue<>(key.key(), value);
            }
        });
        // 把上面的结果放到另一个topic中
        return kStream1;

    }
}
